ValueScan 代幣即時訊號訂閱

生產環境：https://stream.valuescan.ai

一、概述

使用者訂閱指定代幣後，當該代幣產生訊號（機會訊號、風險訊號、資金異動訊號）時，伺服端主動推送訊息到用戶端。

主要特性：

秒級即時推送：訊號產生後立即推送到用戶端
多訊號類型：支援機會訊號（OPPORTUNITY）、風險訊號（RISK）、資金異動訊號（FUNDS）
斷線自動重連：用戶端自動處理連線中斷並重試

二、訂閱位址

GET https://stream.valuescan.ai/stream/signal/subscribe

鑑權參數（Query String）：

參數	類型	必填	說明
`apiKey`	string	是	ValueScan Access Key
`sign`	string	是	HMAC-SHA256 簽名
`timestamp`	long	是	毫秒級 Unix 時間戳
`nonce`	string	是	隨機字串（防重放）
`tokens`	string	是	訂閱的代幣 ID，多個代幣以逗號分隔，如 `1,2,3`，傳空值預設為全部代幣

簽名演算法：

sign = HMAC-SHA256(SK, timestamp 毫秒 + nonce)

範例（Python）：

python

import hashlib
import hmac
import time
import uuid

ts = int(time.time() * 1000)          # 毫秒時間戳
nonce = uuid.uuid4().hex               # 隨機字串
sign = hmac.new(
    SK.encode(),
    (str(ts) + nonce).encode(),
    hashlib.sha256
).hexdigest()

時間戳有效期：±300 秒（5 分鐘），超出返回 401。

三、SSE 事件格式

連線成功

event: connected
data: subscribed

心跳（維持連線）

: heartbeat

event: heartbeat
data: ping

伺服端每 20 秒發送一次心跳。

訊號推送

event: signal
data: {"content":"{...}","tokenId":1,"type":"OPPORTUNITY","ts":1775604300144,"uniqueKey":"1_1775604300144"}

signal 事件 payload 欄位：

欄位	類型	說明
`tokenId`	long	代幣 ID
`type`	string	訊號類型
`content`	string	訊號詳情 JSON 字串
`ts`	long	毫秒時間戳
`uniqueKey`	string	去重識別

四、三種訊號類型

1. 機會訊號（OPPORTUNITY）

業務含義：AI 透過多數據維度，追蹤當前行情下具有上漲潛力的代幣的即時追蹤訊息。

對應查詢介面：POST /open/v1/ai/getChanceCoinMessageList

content 結構：

欄位	類型	說明
`vsTokenId`	string	代幣 ID
`symbol`	string	代幣符號，如 BTC
`name`	string	代幣名稱，如 Bitcoin
`chanceMessageType`	int	機會訊息類型（主力動作類型）
`scoring`	number	機會得分，分數越高訊號越強
`grade`	int	訊號等級
`price`	string	當前價格
`percentChange24h`	number	24 小時漲跌幅
`gains`	number	漲幅
`decline`	number	跌幅
`updateTime`	long	更新時間（毫秒時間戳）

2. 風險訊號（RISK）

業務含義：AI 透過多數據維度，追蹤當前行情下具有下跌風險的代幣的即時追蹤訊息。

對應查詢介面：POST /open/v1/ai/getRiskCoinMessageList

content 結構：

欄位	類型	說明
`vsTokenId`	string	代幣 ID
`symbol`	string	代幣符號
`name`	string	代幣名稱
`riskMessageType`	int	風險訊息類型
`scoring`	number	風險得分（可能為空）
`grade`	int	訊號等級（可能為空）
`price`	string	當前價格
`percentChange24h`	number	24 小時漲跌幅
`riskDecline`	number	風險跌幅（價格下跌幅度）
`rebound`	number	反彈幅度
`updateTime`	long	更新時間（毫秒時間戳）

3. 資金異動訊號（FUNDS）

業務含義：監控中心化交易所的現貨或合約交易中，主力資金異動的即時追蹤訊息。

對應查詢介面：POST /open/v1/ai/getFundsCoinMessageList

content 結構：

欄位	類型	說明
`vsTokenId`	string	代幣 ID
`symbol`	string	代幣符號
`name`	string	代幣名稱
`tradeType`	int	交易類型（1:現貨 2:合約 3:交割合約）
`price`	string	當前價格
`percentChange24h`	number	24 小時漲跌幅
`fundsMovementType`	int	資金流動類型（1:現貨; 2:合約; 等）
`updateTime`	long	更新時間（毫秒時間戳）

五、訂閱範例

Python（推薦）

python

import hashlib
import hmac
import json
import time
import uuid
import urllib.request
from urllib.parse import urlencode

AK = "your-access-key"
SK = "your-secret-key"
TOKENS = "1,2,3"          # 訂閱的代幣 ID

ts = int(time.time() * 1000)
nonce = uuid.uuid4().hex
sign = hmac.new(
    SK.encode(),
    (str(ts) + nonce).encode(),
    hashlib.sha256,
).hexdigest()

params = urlencode({
    "apiKey": AK,
    "sign": sign,
    "timestamp": ts,
    "nonce": nonce,
    "tokens": TOKENS,
})

url = f"https://stream.valuescan.ai/stream/signal/subscribe?{params}"
req = urllib.request.Request(url, headers={"Accept": "text/event-stream"})

with urllib.request.urlopen(req, timeout=300) as resp:
    event_name = ""
    data_str = ""
    for raw in resp:
        line = raw.decode("utf-8").rstrip("\r\n")

        if line.startswith(":"):
            print("♥ 心跳")
            continue

        if line == "":
            if data_str:
                handle_event(event_name, data_str)
            event_name = ""
            data_str = ""
        elif line.startswith("event:"):
            event_name = line[6:].strip()
        elif line.startswith("data:"):
            data_str = line[5:].strip()


def handle_event(event_name: str, data_str: str) -> None:
    if event_name == "signal":
        payload = json.loads(data_str)
        token_id = payload["tokenId"]
        signal_type = payload["type"]
        content = json.loads(payload["content"])
        print(f"[訊號] tokenId={token_id} type={signal_type} content={content}")
    elif event_name == "connected":
        print(f"[連線] {data_str}")

cURL（快速驗證）

bash

# 產生簽名參數
python3 -c "
import hashlib, hmac, time, uuid, urllib.parse
SK='your-secret-key'
ts=int(time.time()*1000)
nonce=uuid.uuid4().hex
sign=hmac.new(SK.encode(), (str(ts)+nonce).encode(), hashlib.sha256).hexdigest()
params=urllib.parse.urlencode({'apiKey':'your-access-key','sign':sign,'timestamp':ts,'nonce':nonce,'tokens':'1,2'})
print(f'curl -N \"https://stream.valuescan.ai/stream/signal/subscribe?{params}\" -H \"Accept: text/event-stream\"')
'"

將輸出的指令貼上執行即可。

六、斷線重連

SSE 連線可能因網絡波動或伺服端重啟而中斷。建議用戶端實作指數退避重連：

python

import time

def listen_with_retry(url: str, max_retries: int = 10) -> None:
    retries = 0
    while retries <= max_retries:
        try:
            req = urllib.request.Request(url, headers={"Accept": "text/event-stream"})
            with urllib.request.urlopen(req, timeout=300) as resp:
                # 正常退出（使用者中斷）則不重連
                _read_sse(resp)
                break
        except KeyboardInterrupt:
            print("使用者中斷，退出。")
            break
        except Exception as e:
            retries += 1
            wait = min(2 ** retries, 60)
            print(f"連線中斷（{e}），{wait}s 後第 {retries} 次重連...")
            time.sleep(wait)
    else:
        print("重試次數耗盡，退出。")

七、內容解析範例

content 欄位是 JSON 字串，需要二次解析：

python

def parse_signal_content(content_str: str) -> dict:
    return json.loads(content_str)

# 範例
payload = json.loads(data_str)
content = parse_signal_content(payload["content"])

if payload["type"] == "OPPORTUNITY":
    print(f"機會訊號：{content['name']} ({content['symbol']}) "
          f"價格={content['price']} 得分={content['scoring']}")
elif payload["type"] == "RISK":
    print(f"風險訊號：{content['name']} ({content['symbol']}) "
          f"價格={content['price']} 風險跌幅={content['riskDecline']}%")
elif payload["type"] == "FUNDS":
    trade = "現貨" if content["tradeType"] == 1 else "合約"
    print(f"資金異動：{content['name']} ({content['symbol']}) "
          f"{trade} 價格={content['price']}")

八、SDK 接入（Python）

推薦使用官方 SDK，只需約 20 行程式碼即可完成訂閱：

python

from agent.sdk.client import SseClient

def on_event(event_name: str, payload: dict) -> None:
    if event_name == "signal":
        import json
        content = json.loads(payload["content"])
        print(f"[{payload['type']}] {content.get('name', '')} "
              f"{content.get('price', '')}")

client = SseClient(
    url="https://stream.valuescan.ai/stream/signal/subscribe?<已簽名的參數>",
    on_event=on_event,
)
client.start()

SDK 內部處理：心跳解析、斷線重連（指數退避）、事件分發。

九、錯誤碼

HTTP 狀態碼	說明
200	連線成功
400	參數缺失或格式錯誤
401	簽名驗證失敗或時間戳超限
404	訂閱位址錯誤
503	伺服端不可用（鑑權後端連線失敗）

ValueScan 代幣即時訊號訂閱 ​

一、概述 ​

二、訂閱位址 ​

三、SSE 事件格式 ​

連線成功 ​

心跳（維持連線） ​

訊號推送 ​

四、三種訊號類型 ​

1. 機會訊號（OPPORTUNITY） ​

2. 風險訊號（RISK） ​

3. 資金異動訊號（FUNDS） ​

五、訂閱範例 ​

Python（推薦） ​

cURL（快速驗證） ​

六、斷線重連 ​

七、內容解析範例 ​

八、SDK 接入（Python） ​

九、錯誤碼 ​

ValueScan 代幣即時訊號訂閱

一、概述

二、訂閱位址

三、SSE 事件格式

連線成功

心跳（維持連線）

訊號推送

四、三種訊號類型

1. 機會訊號（OPPORTUNITY）

2. 風險訊號（RISK）

3. 資金異動訊號（FUNDS）

五、訂閱範例

Python（推薦）

cURL（快速驗證）

六、斷線重連

七、內容解析範例

八、SDK 接入（Python）

九、錯誤碼